package com.dmp.report

import com.dmp.config.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

object PeoCityRrport {
  def main(args: Array[String]): Unit = {
    if (args.length != 1) {
      println(
        """
          |Usage:
          | cn.sheep.violet.etl.Bz2Parquet
          | args:
          |     dataInputPath:  原始日志输入路径
        """.stripMargin)
      sys.exit(101) // 101: 参数不合法 $?
    }
    val Array(inpath) = args
    Logger.getLogger("org").setLevel(Level.WARN)
    val sparkConf = new SparkConf()
    sparkConf.setAppName("地域分布情况")
    sparkConf.setMaster("local[*]")
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    var i=0
    i+=1
    val sc = new SparkContext(sparkConf)
    val sQLContext = new SQLContext(sc)
    val dataFrame: DataFrame = sQLContext.read.parquet(inpath)
    dataFrame.registerTempTable("dmpLog")
    /*val value: RDD[(String, String)] = */
    sQLContext.sql("select provincename,cityname from dmpLog")
      .distinct().registerTempTable("proCity")
    val value: RDD[String] = sQLContext.sql("select provincename from dmpLog")
      .distinct().toDF().map(df => {
      df.getAs[String]("provincename")
    })

    /*    value.foreach(vl=>{
         sQLContext.sql(s"select cityname from proCity where provincename='${vl}'")
            .show()
        })*/
/*      .toDF().map(df => {
      val province: String = df.getAs[String]("provincename")
      val cityname: String = df.getAs[String]("cityname")
      (province, cityname)
    })
    value.foreach(println)*/

      sQLContext.sql("select cityname,count(processnode) request," +
        " count(if (processnode=2,true,null)) validPro," +
        " count(if(processnode=3,true,null)) advicePro," +
        " count(requestmode) resprice, " +
        "count(if(iswin=0,true,null)) winnum," +
        "concat((count(requestmode)/count(if(iswin=0,true,null))*100),'%')LBR," +
        "count(if(requestmode=2,true,null)) shownum ," +
        " count(if(requestmode=3,true,null)) clicknum," +
        " concat((count(if(requestmode=2,true,null))/count(if(requestmode=3,true,null))*100),'%') CTR," +
        s"count(adpayment) AC,count(adprice) AP from dmpLog group by cityname")
        .write.jdbc(ConfigHandler.url, "cityApli", ConfigHandler.prop)
    sc.stop()
  }

}
